package belf.migrate.engine;

import belf.migrate.api.common.JobMode;
import belf.migrate.api.taskconf.JobContext;
import belf.migrate.api.type.LogLevelType;
import belf.migrate.api.type.MigrationObjectType;
import belf.migrate.api.util.PostLog;
import belf.migrate.api.job.Job;
import belf.migrate.engine.job.JobResult;
import belf.migrate.engine.job.cdc.CDCJobRunner;
import belf.migrate.engine.job.evaluate.EvaluateJobRunner;
import belf.migrate.engine.job.execute.ExecuteJobRunner;
import belf.migrate.engine.job.jdbc.JDBCJobRunner;
import belf.migrate.engine.job.sql.SyncSpeicalRowsRunner;
import belf.migrate.engine.job.structure.DropAllStucture;
import belf.migrate.engine.job.structure.StructureJobRunner;
import belf.migrate.engine.job.structure.SyncAllStucture;
import belf.migrate.engine.job.structure.helper.*;
import com.google.gson.JsonObject;
import belf.migrate.engine.job.verify.VerifyJobRunner;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.Map;

import static belf.migrate.api.taskconf.Task.TaskType.SYNC_DATA;
import static belf.migrate.api.job.JobCode.EXECUTING;
import static belf.migrate.api.job.JobCode.SUCCESS;
import static groovy.json.JsonOutput.toJson;

@Slf4j
public class SubmitJob {
    public JsonObject submitJob(Job job) {
        JsonObject retJson = new JsonObject();


        PostLog.sendLogModel(
                job.getJobId(),
                job.getTask().getTaskType().getName(),
                "任务开始！",
                LogLevelType.INFO.getType(),
                EXECUTING,
                new HashMap<>());

        log.info("Submit job: {}", job.getJobId());
        BelfEvn.getInstance().addJob(job);

        // 启动Job并执行
        switch (job.getTask().getTaskType()) {
            case SYNC_STRUCTURE://同步所有结构，包括表、视图、索引、外键、自定义类型、触发器、函数、存储过程等
                retJson = new SyncAllStucture(job).execute().toJson();
                break;
            case DROP_STRUCTURE:
                retJson = new DropAllStucture(job).execute().toJson();
                break;
            case SYNC_DATA:
                JobMode jobMode = job.getTask().getJobMode();
                if(jobMode == JobMode.STREAMING){
                    retJson = new CDCJobRunner(job).execute().toJson();
                } else if (jobMode == JobMode.BATCH) {
                    retJson = new JDBCJobRunner(job).execute().toJson();
                }
                break;
            case CREATE_TABLE:
                retJson = new StructureJobRunner(job, true, true, false).execute().toJson();
                break;
            case DROP_TABLE:
                retJson = new StructureJobRunner(job, false, true, false).execute().toJson();
                break;
            case CREATE_FOREIGNKEY:
                retJson = new StructureJobRunner(job, false, false, true).execute().toJson();
                break;
            case CREATE_VIEW:
                JobResult createView = new CreateViewTool(job).run();
                createView.setTypeName(MigrationObjectType.VIEW.getName());
                retJson = createView.toJson();
                break;
            case DROP_VIEW:
                JobResult dropView = new DropViewTool(job).run();
                dropView.setTypeName(MigrationObjectType.VIEW.getName());
                retJson = dropView.toJson();
                break;
            case CREATE_TYPE:
                JobResult createType = new CreateTypeTool(job).run();
                createType.setTypeName(MigrationObjectType.TYPE.getName());
                retJson = createType.toJson();
                break;
            case DROP_TYPE:
                JobResult dropType = new DropTypeTool(job).run();
                dropType.setTypeName(MigrationObjectType.TYPE.getName());
                retJson = dropType.toJson();
                break;
            case CREATE_SEQ:
                JobResult createSeq = new CreateSeqTool(job).run();
                createSeq.setTypeName(MigrationObjectType.SEQUENCE.getName());
                retJson = createSeq.toJson();
                break;
            case DROP_SEQ:
                JobResult dropSeq = new DropSeqTool(job).run();
                dropSeq.setTypeName(MigrationObjectType.SEQUENCE.getName());
                retJson = dropSeq.toJson();
                break;
            case CREATE_TRIGGER:
                JobResult createTrigger = new CreateTriggerTool(job).run();
                createTrigger.setTypeName(MigrationObjectType.TRIGGER.getName());
                retJson = createTrigger.toJson();
                break;
            case DROP_TRIGGER:
                JobResult dropTrigger = new DropTriggerTool(job).run();
                dropTrigger.setTypeName(MigrationObjectType.TRIGGER.getName());
                retJson = dropTrigger.toJson();
                break;
            case CREATE_FUNCTION:
                JobResult createFunction = new CreateFunctionTool(job).run();
                createFunction.setTypeName(MigrationObjectType.FUNCTION.getName());
                retJson = createFunction.toJson();
                break;
            case DROP_FUNCTION:
                JobResult dropFunction = new DropFunctionTool(job).run();
                dropFunction.setTypeName(MigrationObjectType.FUNCTION.getName());
                retJson = dropFunction.toJson();
                break;
            case CREATE_SYNONYM:
                JobResult createSynonym = new CreateSynonymTool(job).run();
                createSynonym.setTypeName(MigrationObjectType.SYNONYM.getName());
                retJson = createSynonym.toJson();
                break;
            case DROP_SYNONYM:
                JobResult dropSynonym = new DropSynonymTool(job).run();
                dropSynonym.setTypeName(MigrationObjectType.SYNONYM.getName());
                retJson = dropSynonym.toJson();
                break;
            case CREATE_PROCEDURE:
                JobResult createProcedure = new CreateProcedureTool(job).run();
                createProcedure.setTypeName(MigrationObjectType.PROCEDURE.getName());
                retJson = createProcedure.toJson();
                break;
            case DROP_PROCEDURE:
                JobResult dropProcedure = new DropProcedureTool(job).run();
                dropProcedure.setTypeName(MigrationObjectType.PROCEDURE.getName());
                retJson = dropProcedure.toJson();
                break;
            case EVALUATE_MIGRATION:
            case EVALUATE_TABLE:
            case EVALUATE_VIEW:
            case EVALUATE_TRIGGER:
            case EVALUATE_PROCEDURE:
            case EVALUATE_TYPE:
            case EVALUATE_INDEX:
            case EVALUATE_SEQ:
            case EVALUATE_SYNONYM:
            case EVALUATE_FUNC:
                retJson = new EvaluateJobRunner(job).execute().toJson();
                break;
            case VERIFY_MIGRATION:
            case VERIFY_TABLE:
            case VERIFY_TRIGGER:
            case VERIFY_VIEW:
            case VERIFY_SEQ:
            case VERIFY_FUNC:
            case VERIFY_INDEX:
            case VERIFY_SYNONYM:
            case VERIFY_TYPE:
                retJson = new VerifyJobRunner(job).execute().toJson();
                break;
            case SYNC_SPECIAL_ROWS:
                retJson = new SyncSpeicalRowsRunner(job).execute().toJson();
                break;
            case JDBC_SYNC_DATA:
                retJson = new JDBCJobRunner(job).execute().toJson();
                break;
            case EXECUTE_SQL:
                retJson = new ExecuteJobRunner(job).execute().toJson();
                break;
            default:
                break;

        }
        Map<String, Object> progress = null;
        try {
            JobContext jobContext = BelfEvn.getInstance().getJobContextMap().get("" + job.getJobId());
            if (null != jobContext) {
                progress = jobContext.getProgress();
            }
        } catch (Exception e) {
            log.error("get job context error", e);
        }
        if (job.getTask().getTaskType() != SYNC_DATA) { //非流式任务，自动关闭job
            BelfEvn.getInstance().finishJob(job.getJobId());
        }
        if (!SYNC_DATA.equals(job.getTask().getTaskType())) {
            PostLog.sendLogModel(
                    job.getJobId(),
                    job.getTask().getTaskType().getName(),
                    "任务结束！",
                    LogLevelType.INFO.getType(),
                    SUCCESS,
                    progress);
        }
        return retJson;
    }

}
